//package com.allwees.data.dataport.module.product.mq.consume;
//
//import java.io.IOException;
//import java.nio.charset.StandardCharsets;
//import java.util.ArrayList;
//import java.util.List;
//import java.util.concurrent.Callable;
//import java.util.concurrent.ExecutorService;
//import java.util.concurrent.Executors;
//
//import javax.annotation.PostConstruct;
//
//import org.springframework.amqp.core.AcknowledgeMode;
//import org.springframework.amqp.core.Message;
//import org.springframework.amqp.core.MessageProperties;
//import org.springframework.amqp.rabbit.connection.ConnectionFactory;
//import org.springframework.amqp.rabbit.listener.BlockingQueueConsumer;
//import org.springframework.amqp.rabbit.support.ActiveObjectCounter;
//import org.springframework.amqp.rabbit.support.DefaultMessagePropertiesConverter;
//import org.springframework.amqp.rabbit.support.MessagePropertiesConverter;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.context.ApplicationEventPublisher;
//import org.springframework.data.elasticsearch.annotations.Document;
//
//import com.rabbitmq.client.ShutdownSignalException;
//import com.allwees.bs.core.model.product.entity.ProductModel;
//import com.allwees.core.common.util.json.JsonUtil;
//import com.allwees.core.mq.rabbitmq.config.RabbitMqConfig;
//import com.allwees.data.dataport.module.product.doc.ProductSpu;
//import com.allwees.data.dataport.module.product.repository.es.ProductIndexEsRepository;
//
//import cn.hutool.core.thread.ThreadUtil;
//import lombok.extern.slf4j.Slf4j;
//
///**
// * 批量接收MQ消息并更新索引（利用{@link BlockingQueueConsumer}）
// *
// * @author yangylsky
// * @since 2019-11-23
// *
// */
//@Slf4j
////@Component
////@Lazy(false)
//public class BatchUpdateProductIndexListenerV1 {
//	@Autowired
//	private RabbitMqConfig mqConfig;
//
//	@Autowired
//	private ConnectionFactory connectionFactory;
//
//	private static final String QUEUE_NAME = "q.Product.UpdateIndex";
//
//	private static BlockingQueueConsumer blockingQueueConsumer;
//
//	private static final ExecutorService executor = Executors.newFixedThreadPool(1);
//
//	@Autowired
//	private ApplicationEventPublisher applicationEventPublisher;
//
//	@Autowired
//	private ProductIndexEsRepository productIndexEsRepository;
//
//	@PostConstruct
//	private void _init() {
//		try {
//			mqConfig.checkQueue(QUEUE_NAME);
//
//			blockingQueueConsumer = new BlockingQueueConsumer(connectionFactory,
//					messagePropertiesConverter(),
//					new ActiveObjectCounter<>(),
//					AcknowledgeMode.AUTO,
//					false,
//					mqConfig.getBatchSize() * 2,
//					QUEUE_NAME);
//			blockingQueueConsumer.setApplicationEventPublisher(applicationEventPublisher);
//			blockingQueueConsumer.start();
//
//			executor.submit(new Callable<String>() {
//				@Override
//				public String call() throws Exception {
//					execute();
//					return null;
//				}
//			});
//			log.info("BatchUpdateProductIndexListener is ready.");
//		} catch(Exception e) {
//			log.error("BatchUpdateProductIndexListener init failed", e);
//		}
//	}
//
//	private MessagePropertiesConverter messagePropertiesConverter() {
//		MessagePropertiesConverter converter = new DefaultMessagePropertiesConverter();
//		MessageProperties msgProps = new MessageProperties();
//		msgProps.setContentType(MessageProperties.CONTENT_TYPE_JSON);
//		converter.fromMessageProperties(msgProps, StandardCharsets.UTF_8.name());
//		return converter;
//	}
//
//	private void execute() {
//		List<ProductSpu> list = new ArrayList<>();
//		try {
//			while(true) {
//				int dealNum = 0;
//				long batchStartTime = System.currentTimeMillis();
//				try {
//					int i = mqConfig.getBatchSize();
//					while(i-- > 0) {
//						Message msg = blockingQueueConsumer.nextMessage(mqConfig.getBatchTimeout());
//						if(msg != null) {
//							String json = new String(msg.getBody(), StandardCharsets.UTF_8.name());
//							ProductModel product = JsonUtil.fromJson(json, ProductModel.class);
//							list.add(ProductSpu.fromProductModel(product));
//						}
//						if(System.currentTimeMillis() - batchStartTime > mqConfig.getBatchTimeout()) {
//							break;
//						}
//					}
//					if(!list.isEmpty()) {
//						productIndexEsRepository.saveAll(list);
//						dealNum = list.size();
//						list.clear();
//						blockingQueueConsumer.commitIfNecessary(false);
//					} else {
//						//队列中没有拿到数据，sleep 1s
//						ThreadUtil.safeSleep(1000);
//					}
//				} catch(IOException e) {
//					log.error("", e);
//				} catch(InterruptedException e) {
//					log.error("", e);
//					Thread.currentThread().interrupt();
//				} finally {
//					if(dealNum > 0) {
//						Document docAnno =ProductSpu.class.getAnnotation(Document.class);
//						log.info("Save ProductSpu[num: {}] to index[{}] in {}ms", dealNum, docAnno.indexName(), System.currentTimeMillis() - batchStartTime);
//					}
//				}
//			}
//		} catch(ShutdownSignalException e) {
//			log.error("", e);
//			Thread.currentThread().interrupt();
//		}
//	}
//}
